package org.example

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, desc}
import org.apache.spark.sql.types.{DataTypes, IntegerType, StringType, StructField, StructType}

object data_SQL3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
//    读取电影用户信息
val schemaUser = StructType(Seq(
  StructField("id", IntegerType),
  StructField("gender", StringType),
  StructField("age", IntegerType),
  StructField("occupation", IntegerType),
  StructField("zipcode", StringType),
))
    val user = spark.read.option("sep","::").schema(schemaUser)
      .csv("src/main/resources/users.dat")
    user.show(5)
//    1.查询
    user.where("gender='F' and age= 18 ").show(5)
    user.filter("gender='F' and age= 18 ").show(5)
//    替换字段里面的值 udf
    spark.udf.register("replace",(x:String)=>{
      x match {
        case "M" => 0
        case "F" => 1
      }
    })
//    替换列名
    user.selectExpr("id","replace(gender) as sexual","age").show()
//    col() apply()获取指定列
//    user.select(user.col("zipcode")).collect().foreach(println)
//    user.select(user.apply("zipcode")).collect().foreach(println)
//    2.排序
    user.orderBy(desc("id")).show(5)
    user.orderBy(-user("id")).show(5)
    user.sort(desc("id")).show(5)
//    3.分组
    user.groupBy("gender").count().show()
//    4.连接 join(DataFrame,"")
//    练习：读取评分表，并将用户表和评分表连接，查询18岁女生评分为5分的所有电影id

    val schemaRating = StructType(Seq(
      StructField("id", IntegerType),
      StructField("movie_id",IntegerType ),
      StructField("rating", IntegerType),
      StructField("pm",StringType ),
    ))
    val ratingsDF = spark.read
      .option("sep","::").schema(schemaRating)
      .csv("src/main/resources/ratings.dat")
    ratingsDF.show(5)
    val joinedDF = user.join(ratingsDF, "id")
    val resultDF = joinedDF.filter(col("age") === 18 && col("gender") === "F" && col("rating") === 5)
      .select("movie_id","rating")
    resultDF.show()
    sc.stop()
  }

}
